package com.xxxx.transformation_operator;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;
import java.util.List;
import java.util.Random;

/**
 * @program: flink19Test
 * @description:
 * @author: CoreDao
 * @create: 2021-04-01 16:32
 **/

public class AggOperatorMinBy {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //flink为了达到高吞吐，也是通过buffer来提高吞吐量，不是一条数据就计算。
        //有界流
        env.setBufferTimeout(-1);
        env.setParallelism(1);

        List<Tuple3<String, String, Integer>> list = Arrays.asList(
                Tuple3.of("男", "小王", 80),
                Tuple3.of("男", "老孟", 300),
                Tuple3.of("男", "小张", 70),
                Tuple3.of("男", "老王", 50)
        );

        List<Tuple2<String, Integer>> list2 = Arrays.asList(
                Tuple2.of("男", 80),
                Tuple2.of("男", 300),
                Tuple2.of("男", 70),
                Tuple2.of("男", 50)
        );

        DataStreamSource<Tuple3<String, String, Integer>> dataStreamSource = env.fromCollection(list);
//        DataStreamSource<Tuple2<String, Integer>> dataStreamSource1 = env.fromCollection(list2);

//        dataStreamSource.keyBy(0).min(2).print();
        /*
        (男,小王,80)
        (男,小王,80)
        (男,小王,70)
        (男,小王,50)
        */
//        dataStreamSource1.keyBy(0).min(1).print();
        dataStreamSource.keyBy(0).minBy(2).print();
        /*
            (男,小王,80)
            (男,小王,80)
            (男,小张,70)
            (男,老王,50)
        */

        env.execute();

    }
}
